package raft

import (
	"bytes"
	"labrpc"
	"log"
	"math/rand"
	"sort"
	"sync"
	"sync/atomic"
	"time"

	"mit6.824/src/labgob"
)

const (
	NILL = -10
)
const (
	LEADER = iota
	CANDIDATE
	FOLLOWER
)
const (
	Time_Out = iota
	Vote_Done
	Vote_Success
)

func Max(x int, y int) int {
	if x >= y {
		return x
	} else {
		return y
	}
}

func Min(x int, y int) int {
	if x > y {
		return y
	} else {
		return x
	}
}

type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
}

type LogEntry struct {
	Command interface{}
	Term    int
	Index   int
	IsEmpty bool
}

type Raft struct {
	mu        sync.Mutex
	peers     []*labrpc.ClientEnd
	Persister *Persister
	me        int
	dead      int32

	currentTerm int
	votedFor    int
	log         []LogEntry

	commitIndex int // 已提交的最大log编号
	lastApplied int // 最后被应用到状态机的日志条目索引

	nextIndex  []int // 对于每个服务器，下一个需要发送给它们的日志索引
	matchIndex []int // 对于每个服务器，已经复制给它们的最高索引

	State        int // Sever的状态， LEADER，CANDIDATE 和 FOLLOWER
	candidateOut int
	followerOut  int

	appliedLog []LogEntry

	applyMessage chan ApplyMsg

	CTimeOut int
	FTimeOut int
}

func (rf *Raft) GetState() (int, bool) {
	var term int
	var isleader bool
	term = rf.currentTerm
	if rf.State == LEADER {
		isleader = true
	} else {
		isleader = false
	}
	return term, isleader
}

func (rf *Raft) persist() {
	//2C
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.currentTerm)
	e.Encode(rf.votedFor)
	e.Encode(rf.log)
	data := w.Bytes()
	rf.Persister.SaveRaftState(data)
}

func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 {
		return
	}
	//2C
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var currentTerm int
	var votedFor int
	var logs []LogEntry
	if d.Decode(&currentTerm) != nil || d.Decode(&votedFor) != nil || d.Decode(&logs) != nil {
		//fmt.Println("%v decode error", rf.me)
	} else {
		rf.currentTerm = currentTerm
		rf.votedFor = votedFor
		rf.log = logs
	}
}

func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {
	//2D
	return true
}

func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).

}

// RequestVote RPC
type RequestVoteArgs struct {
	Term         int
	CandidateID  int
	LastLogIntex int
	LastLogTerm  int
}

type RequestVoteReply struct {
	Term        int
	VoteGranded bool
}

type AppendEntriesArgs struct {
	Term         int
	LeaderID     int
	PrevLogIntex int //前一条日志的索引
	PrevLogTerm  int //前一条日志的任期
	Entries      []LogEntry
	LeaderCommit int //领导者已经提交的日志索引值
}

type AppendEntrieReply struct {
	Term          int
	Success       bool
	UpdateNextIdx int // 更新请求节点nextIndex[i]
}

func (rf *Raft) LastTerm() int {
	lastTerm := 0
	if len(rf.log) > 0 {
		lastTerm = rf.log[len(rf.log)-1].Term
	} else {
		lastTerm = rf.currentTerm
	}
	return lastTerm
}

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	reply.Term = rf.currentTerm
	reply.VoteGranded = false
	if args.Term >= rf.currentTerm {
		if args.Term > rf.currentTerm {
			rf.currentTerm = args.Term
			rf.State = FOLLOWER
			rf.votedFor = NILL
			rf.candidateOut = 0
			rf.persist()
		}
		rf.mu.Unlock()
		time.Sleep(10 * time.Millisecond)
		rf.mu.Lock()
		if (rf.votedFor == NILL || rf.votedFor == args.CandidateID) && //没有投票对象或者原先投票对象/就是发出请求的对象
			(args.LastLogTerm > rf.LastTerm() || (args.LastLogTerm == rf.LastTerm() && args.LastLogIntex >= len(rf.log))) {
			reply.VoteGranded = true
			rf.votedFor = args.CandidateID
			rf.currentTerm = args.Term
			rf.State = FOLLOWER
			rf.votedFor = args.CandidateID
			rf.followerOut = 0
			rf.candidateOut = 0
			rf.persist()
		}
	}

	return
}

func (rf *Raft) AppendEntry(args *AppendEntriesArgs, reply *AppendEntrieReply) {
	rf.mu.Lock()
	//defer rf.mu.Unlock()
	reply.Term = rf.currentTerm
	reply.Success = false
	reply.UpdateNextIdx = 1

	if args.Term > rf.currentTerm {
		rf.currentTerm = args.Term
		rf.State = FOLLOWER
		rf.votedFor = args.LeaderID
		rf.persist()
		rf.candidateOut = 0
		rf.followerOut = 0
		rf.mu.Unlock()
		return
	} else if args.Term == rf.currentTerm {
		rf.currentTerm = args.Term
		rf.State = FOLLOWER
		rf.votedFor = args.LeaderID
		rf.persist()
		rf.candidateOut = 0
		rf.followerOut = 0
		if len(args.Entries) == 0 {
			if rf.LastTerm() == args.Term {
				if args.LeaderCommit > rf.commitIndex {
					rf.commitIndex = Min(args.LeaderCommit, len(rf.log)-1)
				}
			}
			reply.UpdateNextIdx = rf.lastApplied
			rf.mu.Unlock()
			return
		}

		if len(rf.log)-1 < args.PrevLogIntex || rf.log[args.PrevLogIntex].Term != args.PrevLogTerm || args.PrevLogIntex < rf.lastApplied-1 {
			reply.UpdateNextIdx = rf.lastApplied
			rf.mu.Unlock()
			return
		}

		rf.log = rf.log[:args.PrevLogIntex+1]
		rf.log = append(rf.log, args.Entries...)
		if args.LeaderCommit > rf.commitIndex {
			rf.commitIndex = Min(args.LeaderCommit, len(rf.log)-1)
		}
		reply.UpdateNextIdx = len(rf.log)
		reply.Success = true
		rf.persist()
		rf.mu.Unlock()
		return
	} else {
		reply.Success = false
		rf.mu.Unlock()
		return
	}
}

func (rf *Raft) sendRequestVote(sever int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[sever].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntry(sever int, args *AppendEntriesArgs, reply *AppendEntrieReply) bool {
	ok := rf.peers[sever].Call("Raft.AppendEntry", args, reply)
	return ok
}

func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.killed() || rf.State != LEADER {
		return index, term, isLeader
	}
	logEntry := LogEntry{
		Command: command,
		Term:    rf.currentTerm,
	}
	rf.log = append(rf.log, logEntry)
	index = len(rf.log) - 1
	term = rf.currentTerm
	rf.persist()

	return index, term, isLeader
}

func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) Timer() {
	for {

		time.Sleep(100 * time.Millisecond)
		rf.mu.Lock()
		if rf.killed() {
			rf.mu.Unlock()
			return
		}

		if rf.State == LEADER || rf.State == CANDIDATE {
			rf.followerOut = 0
			rf.mu.Unlock()
			continue
		} else {
			rf.followerOut += 1
			if rf.followerOut >= rf.FTimeOut {
				rf.followerOut = 0
				rf.State = CANDIDATE
				rf.mu.Unlock()
				go rf.BeCandidate()
			} else {
				rf.mu.Unlock()
			}
		}
	}
}

func (rf *Raft) ApplyEntries() {
	for {
		time.Sleep(30 * time.Millisecond)
		if rf.killed() {
			return
		}

		if rf.State == LEADER {
			rf.mu.Lock()
			middleMatch := make([]int, len(rf.matchIndex))
			copy(middleMatch, rf.matchIndex)
			middleMatch[rf.me] = len(rf.log) - 1
			sort.Ints(middleMatch)
			midIndex := len(middleMatch) / 2
			if middleMatch[midIndex] == -1 || middleMatch[midIndex] == rf.commitIndex {
				rf.mu.Unlock()
				continue
			}
			rf.mu.Unlock()
			if len(rf.log) > 0 {
				if rf.log[middleMatch[midIndex]].Term == rf.currentTerm {
					rf.commitIndex = Max(rf.commitIndex, middleMatch[midIndex]) //有多数follower匹配到log，提交到当前匹配到log编号或者已经提交的最大值
					log.Printf("Leader: %d update commit: %d", rf.me, rf.commitIndex)

					rf.SendHeartBeat() //心跳已经包含当前提交的index
				}
			}
		}
		rf.mu.Lock()
		for rf.lastApplied <= rf.commitIndex {

			if rf.log[rf.lastApplied].IsEmpty == true {
				rf.lastApplied++
				continue
			} else {
				message := ApplyMsg{true, rf.log[rf.lastApplied].Command, rf.lastApplied}
				rf.applyMessage <- message
				rf.appliedLog = append(rf.appliedLog, rf.log[rf.lastApplied])
			}
			rf.lastApplied++
		}
		rf.mu.Unlock()
	}
}

func (rf *Raft) SendHeartBeat() {
	for peer := 0; peer < len(rf.peers); peer++ {
		if peer == rf.me {
			continue
		}
		args := &AppendEntriesArgs{
			Term:         rf.currentTerm,
			LeaderID:     rf.me,
			PrevLogIntex: len(rf.log),
			PrevLogTerm:  rf.LastTerm(),
			Entries:      make([]LogEntry, 0),
			LeaderCommit: rf.commitIndex,
		}
		reply := &AppendEntrieReply{
			Term:          0,
			Success:       false,
			UpdateNextIdx: 0,
		}
		go func(sever int) {
			rf.sendAppendEntry(sever, args, reply)
		}(peer)
	}
}

func (rf *Raft) SyncAppendEntry(peer int) {
	for {
		if rf.killed() {
			return
		}
		for {
			rf.mu.Lock()
			if rf.State != LEADER {
				rf.mu.Unlock()
				return
			}
			if rf.nextIndex[peer] >= len(rf.log) {
				args := &AppendEntriesArgs{
					Term:         rf.currentTerm,
					LeaderID:     rf.me,
					PrevLogIntex: len(rf.log),
					PrevLogTerm:  rf.LastTerm(),
					Entries:      make([]LogEntry, 0),
					LeaderCommit: rf.commitIndex,
				}
				reply := &AppendEntrieReply{
					Term:          0,
					Success:       false,
					UpdateNextIdx: 0,
				}
				rf.mu.Unlock()
				append_done := make(chan bool)
				ok := false
				go func(sever int) {
					ok = rf.sendAppendEntry(sever, args, reply)
					append_done <- true
				}(peer)
				select {
				case <-append_done:
				}
				if ok {
					rf.mu.Lock()
					if reply.Term <= rf.currentTerm {
						if len(args.Entries) == 0 && rf.nextIndex[peer] > 0 {
							rf.nextIndex[peer] = Max(reply.UpdateNextIdx, rf.matchIndex[peer]+1)
						}
						rf.mu.Unlock()
					} else {
						rf.mu.Unlock()
						rf.Befollower(reply.Term, NILL)
						return
					}

				}
				break
			}

			var data = make([]LogEntry, len(rf.log)-rf.nextIndex[peer])

			if len(data) > 0 {
				copy(data, rf.log[rf.nextIndex[peer]:len(rf.log)])
			}
			args := &AppendEntriesArgs{
				Term:         rf.currentTerm,
				LeaderID:     rf.me,
				PrevLogIntex: Max(rf.nextIndex[peer]-1, 0),
				PrevLogTerm:  rf.log[rf.nextIndex[peer]-1].Term,
				Entries:      data,
				LeaderCommit: rf.commitIndex,
			}
			reply := &AppendEntrieReply{
				Term:          0,
				Success:       false,
				UpdateNextIdx: 0,
			}
			rf.mu.Unlock()

			time_out := make(chan bool)
			append_done := make(chan bool)
			ok := false
			go func(sever int) {
				ok = rf.sendAppendEntry(sever, args, reply)
				append_done <- true
			}(peer)

			select {
			case <-time_out:
				//log.Println("timeout")
			case <-append_done:
				//log.Println("append_done")
			}

			if ok {
				rf.mu.Lock()
				if reply.Success {
					rf.nextIndex[peer] += len(data)
					rf.matchIndex[peer] = Max(rf.matchIndex[peer], rf.nextIndex[peer]-1)
					rf.mu.Unlock()
				} else {
					if reply.Term <= rf.currentTerm {
						if len(args.Entries) != 0 && rf.nextIndex[peer] > 0 { //发送的是日志，日志匹配失败
							rf.nextIndex[peer] = reply.UpdateNextIdx
						}
						rf.mu.Unlock()
					} else { //本服务器的任期小于目标服务器的任期
						rf.mu.Unlock()
						rf.Befollower(reply.Term, NILL)
						return
					}
				}

			} else {
				break
			}
		}
		time.Sleep(50 * time.Millisecond)
	}
}

func (rf *Raft) Befollower(term int, voteFor int) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	rf.currentTerm = term
	rf.State = FOLLOWER
	rf.votedFor = voteFor
	rf.candidateOut = 0
	rf.persist()
}

func (rf *Raft) BeLeader() {
	rf.mu.Lock()

	rf.State = LEADER
	rf.votedFor = rf.me
	rf.candidateOut = 0
	rf.followerOut = 0
	//rf.persist()
	for i := 0; i < len(rf.peers); i++ {
		rf.nextIndex[i] = len(rf.log)
		rf.matchIndex[i] = 0
	}
	rf.mu.Unlock()
	rf.SendHeartBeat()
	for i := 0; i < len(rf.peers); i++ {
		if i == rf.me {
			continue
		}
		go rf.SyncAppendEntry(i)
	}
}

func (rf *Raft) BeCandidate() {
	for {
		if rf.killed() {
			return
		}
		rf.mu.Lock()
		if rf.State != CANDIDATE {
			rf.mu.Unlock()
			return
		}

		rf.currentTerm += 1
		rf.votedFor = rf.me
		//rf.persist()
		rf.followerOut = 0
		voteForMe := 1
		//Term := rf.currentTerm
		rf.mu.Unlock()
		waitFlag := sync.WaitGroup{}
		for peer := 0; peer < len(rf.peers); peer++ {
			if peer == rf.me {
				continue
			} else {
				args := &RequestVoteArgs{
					Term:         rf.currentTerm,
					CandidateID:  rf.me,
					LastLogIntex: len(rf.log),
					LastLogTerm:  rf.LastTerm(),
				}
				reply := &RequestVoteReply{
					Term:        rf.currentTerm,
					VoteGranded: false,
				}
				waitFlag.Add(1)
				go func(sever int) {
					ok := rf.sendRequestVote(sever, args, reply)

					rf.mu.Lock()
					if ok {
						if reply.VoteGranded {
							voteForMe++
						} else {
							if reply.Term > rf.currentTerm {
								rf.mu.Unlock()
								rf.Befollower(reply.Term, NILL)
								return
							}
						}
					}
					waitFlag.Done()
					rf.mu.Unlock()
				}(peer)

			}
		}

		stop_flag := make(chan bool)

		// 检查是否超时
		go func() {
			time.Sleep(time.Duration(rf.CTimeOut*100) * time.Millisecond)
			stop_flag <- true
		}()
		// 检查是否所有requestVote都返回
		go func() {
			waitFlag.Wait()
			stop_flag <- true

		}()
		// 检查是否已经可以成为Leader
		go func() {
			for readtime := 0; readtime < rf.CTimeOut; readtime++ {
				time.Sleep(100 * time.Millisecond)
				if voteForMe*2 >= len(rf.peers) {
					stop_flag <- true
				}
			}
		}()

		select {
		case <-stop_flag:
		}

		rf.mu.Lock()
		if voteForMe*2 >= len(rf.peers) {
			rf.mu.Unlock()
			go rf.BeLeader()
			return
		} else {
			rf.mu.Unlock()
		}
		time.Sleep(time.Duration(150+rand.Intn(150)) * time.Millisecond)
	}
}

func Make(peers []*labrpc.ClientEnd, me int, persister *Persister, applych chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.Persister = persister
	rf.me = me
	rf.currentTerm = 0
	rf.votedFor = NILL
	rf.log = make([]LogEntry, 0)
	rf.log = append(rf.log, LogEntry{0, -1, 0, true})
	rf.appliedLog = make([]LogEntry, 0)
	rf.commitIndex = 0
	rf.lastApplied = 1
	rf.CTimeOut = 4 + rand.Intn(4)
	rf.FTimeOut = 6 + rand.Intn(4)

	rf.nextIndex = make([]int, len(peers))
	rf.matchIndex = make([]int, len(peers))
	for i := 0; i < len(peers); i++ {
		rf.nextIndex[i] = 0
		rf.matchIndex[i] = 0
	}
	rf.State = FOLLOWER
	rf.candidateOut = 0
	rf.followerOut = 0
	rf.applyMessage = applych
	rf.readPersist(persister.ReadRaftState())
	go rf.Timer()
	go rf.ApplyEntries()
	return rf
}
